package com.gin.monitor.agg.distribution;

import com.gin.monitor.agg.utils.JobEnvUtils;
import com.gin.monitor.agg.vo.TrafficInfo;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.HashSet;


/**
 * 违法车辆分析
 *
1620637885000,0002,75312,京L22188,190.1,27,31
1620637886000,0002,75312,京L22188,190.1,27,31
1620637889000,0002,75312,京L22188,190.1,27,31
1620637999000,0002,75312,京L22189,190.1,27,31
 *
 * @author gin
 * @date 2021/4/30
 */
public class AreaDistributionAnalysis {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = JobEnvUtils.initEnv(args);

        //第一种，当前的违法车辆（在5分钟内）如果已经出警了。（最后输出道主流中做删除处理）。
        // nc -lk 8888
        SingleOutputStreamOperator<TrafficInfo> stream = env.socketTextStream("node01", 8888)
                .map(new RichMapFunction<String, TrafficInfo>() {
                    @Override
                    public TrafficInfo map(String line) {
                        String[] s = line.split(",");
                        return new TrafficInfo(Long.parseLong(s[0]), s[1], s[2], s[3], Double.parseDouble(s[4]), s[5], s[6]);
                    }
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<TrafficInfo>forBoundedOutOfOrderness(Duration.ofSeconds(20))
                        .withTimestampAssigner((event, timestamp) -> event.actionTime));


        stream.keyBy((KeySelector<TrafficInfo, String>) trafficInfo -> trafficInfo.areaId)
        .timeWindow(Time.seconds(10))
        .apply(new RichWindowFunction<TrafficInfo, String, String, TimeWindow>() {

            HashSet<String> carSet = new HashSet<>();

            @Override
            public void apply(String k, TimeWindow window, Iterable<TrafficInfo> input, Collector<String> out) throws Exception {
                for (TrafficInfo trafficInfo : input) {
                    carSet.add(trafficInfo.car);
                }
                out.collect("区域" + k + ",在窗口期开始时间" + window.getStart()
                        + ",到窗口结束时间" + window.getStart() + " ,一共有: " + carSet.size() + " 辆车");
            }
        })
        .print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

}
